package com.sand.pro.flinksql_pro.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Kafka2KafkaUpset {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 注册SourceTable: source_sensor
        tableEnv.executeSql("create table source_sensor (id string, ts bigint, vc int) with("
                + "'connector' = 'kafka',"
                + "'topic' = 'topic_source_sensor',"
                + "'properties.bootstrap.servers' = '172.28.246.66:9092,172.28.246.67:9092,172.28.246.68:9092',"
                + "'properties.group.id' = 'bigdata',"
                + "'scan.startup.mode' = 'earliest-offset',"
                + "'format' = 'json'"
                + ")");

        // 2. 注册SinkTable: sink_sensor
        tableEnv.executeSql("create table sink_sensor(id string, ts bigint, vc int,PRIMARY KEY(id) NOT ENFORCED) with("
                + "'connector' = 'upsert-kafka',"
                + "'topic' = 'topic_sink_sensor_upset_2',"
                + "'properties.bootstrap.servers' = '172.28.246.66:9092,172.28.246.67:9092,172.28.246.68:9092',"
                + "'properties.group.id' = 'user_log',"
                + "'key.format' = 'json',"
                + " 'key.json.ignore-parse-errors' = 'true',"
                + "'value.format' = 'json',"
                + "'value.json.fail-on-missing-field' = 'false',"
                + "'value.fields-include' = 'ALL'"
                + ")");

        // 3. 从SourceTable 查询数据, 并写入到 SinkTable
        tableEnv.executeSql("insert into sink_sensor select * from source_sensor");
    }

}
